package com.ng.biz.service.impl;

import java.util.Date;
import java.util.Iterator;
import java.util.List;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentMap;
import java.util.concurrent.atomic.AtomicInteger;

import org.eclipse.paho.client.mqttv3.MqttException;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;
import com.alibaba.fastjson.JSONObject;
import com.ng.biz.model.domain.MqttClient;
import com.ng.biz.service.IDataErrorLog;
import com.ng.biz.service.IDataFlow;
import com.ng.biz.service.IMqttClient;
import com.ng.common.component.mqtt.LDMPMqttClient;
import com.ng.common.model.response.R;

import lombok.extern.slf4j.Slf4j;
@Slf4j
@Service
public class MqttClientImpl implements IMqttClient{
	
	@Autowired
	private IDataFlow dataFlowService;
	
    private final ConcurrentMap<String, LDMPMqttClient> mqttClientPool = new ConcurrentHashMap<>();
    
    private final AtomicInteger messageCount = new AtomicInteger(0);
    
    private final AtomicInteger publishCount = new AtomicInteger(0);
    
    
	@Autowired
	private IDataErrorLog dataErrorLogService;
    


	@Override
	public boolean containsKey(String taskId) {
		return mqttClientPool.containsKey(taskId);
	}

	@Override
	public void stop(String id) {
		if(containsKey(id)) {
			mqttClientPool.get(id).disconnect(); 
			mqttClientPool.remove(id);
		}
	}

	@Override
	public R<String> start(String mqttId) {
		MqttClient mqttClient=MqttClient.dao.findById(mqttId);
		
		if(mqttClient==null) {
			return R.failed("MQTT数据不存在.");
		}
		LDMPMqttClient ldmpMqttClient=new  LDMPMqttClient(mqttClient.getId(),mqttClient.getHost(), mqttClient.getUsername(), mqttClient.getPassword(), mqttClient.getPrefix(),mqttClient, mqttClient.getTimeout(), mqttClient.getKeepalive());
		try {
			if( ldmpMqttClient.connect()) {
				mqttClientPool.put(mqttClient.getId(), ldmpMqttClient);
				mqttClient.setLastTime(new Date(System.currentTimeMillis()));
				mqttClient.update();
				return R.ok();
			}else {
				return R.failed();
			}
		} catch (MqttException e) {
			e.printStackTrace();
			log.error("MQTT客户端启动异常:{}",e);
			return R.failed(e.getMessage());
		}
	}

	@Override
	public void executeFlow(String clientId,String topic, JSONObject topicObject, String payload) {
		try {
			dataFlowService.execute("","",topicObject.getJSONObject("flow"), payload);
		} catch (Exception e) {
			log.error("executeFlow异常:{}",e);
			dataErrorLogService.insertOrUpdateErrorLog(clientId, topicObject.getString("name"), "MQTT", "error", e.getMessage());
		}
	}

	@Override
	public void publish(String clientId,String topic, String payload, int qos, boolean retained) {
		if(mqttClientPool.containsKey(clientId)) {
			mqttClientPool.get(clientId).publish(topic, payload, qos, retained);
		}
	}

	@Override
	public void startDefaultBoot() {
    	List<MqttClient> mqttClients=MqttClient.dao.find("select * from "+MqttClient.TABLE_NAME+" where default_running='1' and enabled='1'");
    	for(MqttClient mqttClient:mqttClients) {
    		LDMPMqttClient ldmpMqttClient=new  LDMPMqttClient(mqttClient.getId(),mqttClient.getHost(), mqttClient.getUsername(), mqttClient.getPassword(), mqttClient.getPrefix(),mqttClient, mqttClient.getTimeout(), mqttClient.getKeepalive());
    		try {
    			if( ldmpMqttClient.connect()) {
    				mqttClientPool.put(mqttClient.getId(), ldmpMqttClient);
    				mqttClient.setLastTime(new Date(System.currentTimeMillis()));
    				mqttClient.update();
    			}
    		} catch (MqttException e) {
    			log.error("MQTT客户端启动异常:{}",e);
    		}
    	}		
	}

	@Override
	public void stopAllBoot() {
		Iterator<String> it=mqttClientPool.keySet().iterator();
		String key="";
		while(it.hasNext()) {
			key=it.next();
			mqttClientPool.get(key).disconnect(); 
		}
		mqttClientPool.clear();
	}

	@Override
	public void incrementAndGet() {
		messageCount.incrementAndGet();
	}

	@Override
	public int getAndSet(int delta) {
		return messageCount.addAndGet(delta);
	}

	@Override
	public void useSysPublish(String topic, String payload, int qos, boolean retained) {
		MqttClient mqttClient=MqttClient.dao.findFirst("select * from biz_mqtt_client a where  a.use='0' and (a.type='1' OR a.type='2') ");
		if(mqttClient!=null) {
			 publish(mqttClient.getId(), topic,  payload,  qos,  retained);
		}
	}

	@Override
	public int getSendAndSet(int delta) {
		return publishCount.addAndGet(delta);
	}

	@Override
	public void sendIncrementAndGet() {
		publishCount.incrementAndGet();
	}
}
